#include "udp_client.h"
#include "../dns_resolver.h"
#include "../fd_channel.h"
#include "../libevent.h"

namespace evnetplus {

namespace udp {

UDPClient::UDPClient(EventLoop* loop,
                     const std::string& remote_addr/*host:port*/,
                     const std::string& name)
    : status_(kNotCreated)
    , loop_(loop)
    , remote_addr_(remote_addr)
    , name_(name)
    , sockfd_(INVALID_SOCKET)
    , message_callback_fn_(nullptr)
    , message_dotted_callback_fn_(nullptr) {
    if (sock::SplitHostPort(remote_addr.data(), remote_host_, remote_port_)) {
        raddr_ = sock::ParseFromIPPort(remote_addr_.data());
    }
}

UDPClient::~UDPClient() {
    EVNP_DLOG_TRACE;
    assert(loop_->IsInLoopThread());
    if (status_ == kDNSResolving) {
        assert(!read_channel_.get());
        assert(!dns_resolver_.get());
    } else if (status_ != kNotCreated) {
        EVNP_DLOG_TRACE << "close(" << read_channel_->fd() << ")";
        assert(read_channel_->fd() == sockfd_);
        read_channel_->Close();
        read_channel_.reset();

        EVUTIL_CLOSESOCKET(sockfd_);
        sockfd_ = INVALID_SOCKET;
    }
}

void UDPClient::Create() {
    EVNP_DLOG_TRACE << remote_addr_ << " status=" << StatusToString();
    assert(loop_->IsInLoopThread());

    if (!sock::IsZeroAddress(&raddr_)) {
        CreateSocket();
        return;
    }

    status_ = kDNSResolving;
    auto f = std::bind(&UDPClient::OnDNSResolved, shared_from_this(), std::placeholders::_1);
    dns_resolver_ = std::make_shared<DNSResolver>(loop_, remote_host_, creating_timeout_, f);
    dns_resolver_->Start();
}

void UDPClient::Destroy() {
    EVNP_DLOG_TRACE << "Destroy client " << remote_addr_ << " status=" << StatusToString();
    assert(loop_->IsInLoopThread());
    if (dns_resolver_) {
        dns_resolver_->Cancel();
        dns_resolver_.reset();
    }

    if (status_ == kDNSResolving) {
        assert(read_channel_.get() == nullptr);
    }

    if (read_channel_.get()) {
        assert(status_ != kDNSResolving);
        read_channel_->DisableAllEvent();
        read_channel_->Close();
    }
}

void UDPClient::SetMessageCallback(const UDPClient::MessageCallback& cb) {
    message_callback_fn_ = cb;
}

void UDPClient::SetMessageCallback(const UDPClient::MessageDottedCallback& cb) {
    message_dotted_callback_fn_ = cb;
}

void UDPClient::SendMsg(const slice& message) {
    output_buffer_.Append(message);
    write_watcher_->Notify();
}

void UDPClient::onReceivedHandle()
{
    size_t recv_buf_size = 65536U;

    MessagePtr recv_msg(new Message(sockfd_, recv_buf_size));
    socklen_t addr_len = sizeof(struct sockaddr);
    int readn = ::recvfrom(sockfd_, (char*)recv_msg->WriteBegin(), recv_buf_size, 0, recv_msg->mutable_remote_addr(), &addr_len);
    if (readn >= 0) {
        recv_msg->WriteBytes(readn);
        if (message_callback_fn_) {
            message_callback_fn_(recv_msg);
        }

        if (message_dotted_callback_fn_) {
            message_dotted_callback_fn_(recv_msg->remote_ip(),
                    ntohs(((struct sockaddr_in*)recv_msg->remote_addr())->sin_port), (Buffer*)&recv_msg);
        }
    }
}

void UDPClient::onSendHandle() {
    if (output_buffer_.length() == 0) {
        return;
    }

    std::string ss = output_buffer_.NextAllString();

    bool sb = SendMessage(sockfd_, (sockaddr*)&raddr_, ss);
    if (sb) {
        // success
    } else {
        // failed
    }
}

void UDPClient::OnDNSResolved(const std::vector <struct in_addr>& addrs) {
    EVNP_DLOG_TRACE << "addrs.size=" << addrs.size() << " this=" << this;
    if (addrs.empty()) {
        EVNP_LOG_ERROR << "this=" << this << " DNS Resolve failed. host=" << dns_resolver_->host();
        HandleError();
        return;
    }

    struct sockaddr_in* addr = sock::sockaddr_in_cast(&raddr_);
    addr->sin_family = AF_INET;
    addr->sin_port = htons(remote_port_);
    addr->sin_addr = addrs[0];
    status_ = kDNSResolved;

    CreateSocket();
}

void UDPClient::CreateSocket() {
    int serrno = 0;

    /* Create nonblocking socket */
    sockfd_ = ::socket(AF_INET, SOCK_DGRAM, 0);
    if (sockfd_ == INVALID_SOCKET) {
        serrno = errno;
        EVNP_LOG_ERROR << "socket error " << strerror(serrno);
        return;
    }

    if (evutil_make_socket_nonblocking(sockfd_) < 0) {
        goto out;
    }

    sock::SetReuseAddr(sockfd_);
    sock::SetReusePort(sockfd_);

    read_watcher_.reset(new PipeEventWatcher(loop_, [this]{
        onReceivedHandle();
    }));
    read_watcher_->Init();
    read_watcher_->AsyncWait();

    write_watcher_.reset(new PipeEventWatcher(loop_, [this]{
        onSendHandle();
    }));
    write_watcher_->Init();
    write_watcher_->AsyncWait();

    read_channel_.reset(new FdChannel(loop_, sockfd_, true, false));
    read_channel_->SetReadCallback([this]{
        read_watcher_->Notify();
    });
    read_channel_->AttachToLoop();

    status_ = kCreated;

    return;

out:
    EVUTIL_CLOSESOCKET(sockfd_);
    sockfd_ = INVALID_SOCKET;
    status_ = kNotCreated;
}

void UDPClient::HandleError() {
    EVNP_DLOG_TRACE << remote_addr_ << " status=" << StatusToString();
    assert(loop_->IsInLoopThread());
    int serrno = errno;

    // In this error handling method, we will invoke 'conn_fn_' callback function
    // to notify the user application layer in which the user maybe call TCPClient::Disconnect.
    // TCPClient::Disconnect may cause this Connector object desctruct.
    auto self = shared_from_this();

    EVNP_LOG_ERROR << "this=" << this << " status=" << StatusToString() << " fd=" << sockfd_  << " use_count=" << self.use_count() << " errno=" << serrno << " " << strerror(serrno);

    status_ = kNotCreated;

    if (read_channel_) {
        assert(read_channel_ > 0);
        read_channel_->DisableAllEvent();
        read_channel_->Close();
    }

    // Avoid DNSResolver callback again when timeout
    if (dns_resolver_) {
        dns_resolver_->Cancel();
        dns_resolver_.reset();
    }

    // If the connection is refused or it will not try again,
    // We need to notify the user layer that the connection established failed.
    // Otherwise we will try to do reconnection silently.
    if (EVUTIL_ERR_CONNECT_REFUSED(serrno)) {
        // conn_fn_(-1, "");
    }
}

std::string UDPClient::StatusToString() const {
    H_CASE_STRING_BIGIN(status_);
    H_CASE_STRING(kNotCreated);
    H_CASE_STRING(kDNSResolving);
    H_CASE_STRING(kDNSResolved);
    H_CASE_STRING(kCreated);
    H_CASE_STRING_END();
}

}

}
